﻿// Benn Eichhorn 2009
using System;
using System.Text;
using System.Net.Sockets;
using System.IO;
using System.Threading;
using System.Collections.Concurrent;

namespace Algobox.Feeds.Finance.Chix.CHIXMD.Comms
{
    internal class Socket : IDisposable
    {
        internal Socket(ChiXMDClient client)
        {
            m_ChiXMDClient = client;
        }

        /// <summary>
        /// Message termination character
        /// </summary>
        const char TERMINATOR = '\x0A';

        /// <summary>
        /// Heartbeat message
        /// </summary>
        const string HEARTBEAT = "R\x0A";

        /// <summary>
        /// Logout message
        /// </summary>
        const string LOGOUT = "O\x0A";

        /// <summary>
        /// Login message format
        /// </summary>
        const string FORMAT_LOGIN = "L{0,-6}{1,-10}{2,-10}{3,10}\x0A";

        /// <summary>
        /// Message encoding
        /// </summary>
        readonly Encoding ENCODING = Encoding.ASCII;

        const int BUFFERSIZE = 1024;

        /// <summary>
        /// Calling instance
        /// </summary>
        ChiXMDClient m_ChiXMDClient;

        /// <summary>
        /// Underlying TCP socket
        /// </summary>
        TcpClient m_TcpClient;

        /// <summary>
        /// The base network stream
        /// </summary>
        NetworkStream m_NetworkStream;


        ConcurrentQueue<string> m_MessageQueue;

        /// <summary>
        /// Initialise client connection
        /// </summary>
        /// <param name="HostName">Hostname or IP to connect to</param>
        /// <param name="HostPort">Port to connect to</param>
        /// <returns>True if connection was successful</returns>
        internal bool Initialise(string HostName, int HostPort)
        {
            try
            {
                m_TcpClient = new TcpClient(HostName, HostPort);
                m_TcpClient.NoDelay = true;

                if (m_TcpClient.Connected)
                {
                    m_NetworkStream = m_TcpClient.GetStream();

                    m_ChiXMDClient.ProcessState(ChiXMDClient.States.Connected);
                }
            }
            catch
            {
                m_ChiXMDClient.ProcessState(ChiXMDClient.States.FailedToConnect);
                DeInitialise();
            }

            return m_TcpClient.Connected;
        }



        /// <summary>
        /// Closes connection and cleans up threads
        /// </summary>
        /// <returns>True if connection is closed</returns>
        internal bool DeInitialise()
        {
            if (m_ChiXMDClient.IsConnected)
            {
                // Send disconnect event
                m_ChiXMDClient.ProcessState(ChiXMDClient.States.Disconnected);

                // Stop heartbeats
                if (m_TimerHeartbeat != null)
                {
                    m_TimerHeartbeat.Dispose();
                    m_TimerHeartbeat = null;
                }

                // Disconnect
                if (m_TcpClient.Connected)
                {
                    m_TcpClient.Close();
                }

                // Kill marshal
                if (m_ThreadReset != null)
                {
                    m_ThreadReset.Dispose();
                    m_ThreadReset = null;
                }

                // Stop message pump thread
                if (m_ThreadPump != null && m_ThreadPump.IsAlive)
                {
                    m_ThreadPump.Abort();
                    m_ThreadPump.Join();
                    m_ThreadPump = null;
                }

                // Stop socket read thread
                if (m_ThreadRead != null && m_ThreadRead.IsAlive)
                {
                    m_ThreadRead.Abort();
                    m_ThreadRead.Join();
                    m_ThreadRead = null;
                }
            }

            return true;
        }


        public void Dispose()
        {
            DeInitialise();
        }


        /// <summary>
        /// Send a message string to the socket stream
        /// </summary>
        /// <param name="message">Message to be sent</param>
        /// <returns>True if sent successfully</returns>
        private bool SendMessage(string message)
        {
            try
            {
                // encode ASCII to Bytes                
                byte[] bytes = new System.Text.ASCIIEncoding().GetBytes(message);

                // allow only one writer at a time
                lock (this)
                {
                    // send synchronously
                    m_NetworkStream.Write(bytes, 0, bytes.Length);
                }
            }
            catch
            {
                m_ChiXMDClient.ProcessState(ChiXMDClient.States.FailedToSend);
                DeInitialise();
                return false;
            }

            // success
            return true;
        }


        /// <summary>
        /// Attempt to log in to Chi-X
        /// </summary>
        /// <param name="userName">Username</param>
        /// <param name="password">Password</param>
        /// <param name="session">Requested session ID. For initial login leaving this field blank is recommended. For subsequent logins inclusion of the session ID is recommended.</param>
        /// <param name="sequence">Sequence you wish feed to start from (next expected). 1 indicates the beginning of the day. 0 indicates the last message generated by the system bypassing recovery.</param>
        /// <param name="result">Output message describing result of the login attempt</param>
        /// <returns>True on successful login</returns>
        internal bool Login(string userName, string password, string session, int sequence, ref Messages.IMessage result)
        {
            string loginMessage = String.Format(FORMAT_LOGIN, userName, password, session, sequence);

            if (SendMessage(loginMessage))
            {
                StreamReader reader = new StreamReader(m_NetworkStream, Encoding.ASCII);

                string inbound;

                try
                {
                    while (!reader.EndOfStream)
                    {
                        try
                        {
                            inbound = reader.ReadLine();
                        }
                        catch
                        {
                            break;
                        }

                        MessageParser.TryParseSessionMessage(ref inbound, ref result);

                        if (result is Messages.LoginAccept)
                            return true;

                        if (result is Messages.LoginReject || result is Messages.Error)                                                  
                            return false;                        
                    }

                    result = Messages.Error.GetInstance("Failed to recieve inbound message from socket");
                }
                catch (Exception exception)
                {
                    result = Messages.Error.GetInstance("Failed to connect to host: " + exception.Message);
                }
            }
            else
            {
                result = Messages.Error.GetInstance("Failed to send login");
            }

            return false;
        }


        /// <summary>
        /// Logout and disconnect form the Chi-X Market Data system
        /// </summary>
        internal void Logout()
        {
            m_ChiXMDClient.ProcessState(ChiXMDClient.States.Logout);
            
            // send logout
            if(m_ChiXMDClient.IsLoggedIn)
                SendMessage(LOGOUT);

            // message pump and heartbeat timer will be stoped by DeInitialise when the connection becomes disconnected
        }


        /// <summary>
        /// Socket reader thread
        /// </summary>
        Thread m_ThreadRead;

        /// <summary>
        /// Message pump thread
        /// </summary>
        Thread m_ThreadPump;

        /// <summary>
        /// Used to wake up message pump when there is new data to be processed in the queue
        /// </summary>
        AutoResetEvent m_ThreadReset;

        /// <summary>
        /// Start the heartbeat and message processing threads
        /// </summary>
        internal void StartProcessing()
        {
            // heartbeat timer delegate
            m_TimerHeartbeatDelegate = new TimerCallback(SendHeartbeat);

            // heartbeat timer
            m_TimerHeartbeat = new Timer(m_TimerHeartbeatDelegate, null, m_ChiXMDClient.HeartbeatInterval, m_ChiXMDClient.HeartbeatInterval);

            // create message queue
            m_MessageQueue = new ConcurrentQueue<string>();

            // create socket read thread
            m_ThreadRead = new Thread(new ThreadStart(ReadMessages));
            m_ThreadRead.Name = "ChiXMDSocketReader";
            
            // create message pump thread
            m_ThreadPump = new Thread(new ThreadStart(PumpMessages));
            m_ThreadPump.Name = "ChiXMDMessagePump";

            // create reset event handler
            m_ThreadReset = new AutoResetEvent(false);

            // start threads
            m_ThreadRead.Start();
            m_ThreadPump.Start();

        }

        /// <summary>
        /// Heartbeat timer used to send heartbeat messages at specified intervals
        /// </summary>
        Timer m_TimerHeartbeat;

        /// <summary>
        /// Heartbeat timer delagate
        /// </summary>
        TimerCallback m_TimerHeartbeatDelegate;

        /// <summary>
        /// Send a heartbeat message to the socket
        /// Stops timer if sending of heartbeat fails
        /// </summary>
        /// <param name="state">null - required by delegate</param>
        private void SendHeartbeat(Object state)
        {
            if (m_TcpClient.Connected)
            {
                SendMessage(HEARTBEAT);
            }
        }


        /// <summary>
        /// Socket reader thread which reads bytes from the buffer and extracts message strings to place on the message queue
        /// </summary>
        private void ReadMessages()
        {
            byte[] bytes = new byte[BUFFERSIZE];
            
            ASCIIEncoding encoder = new ASCIIEncoding();

            string message = String.Empty;

            int readLength;
            int messageLength;

            try
            {
                while (m_ChiXMDClient.IsConnected)
                {
                    // read from the socket in the bytes buffer
                    readLength = m_NetworkStream.Read(bytes, 0, BUFFERSIZE);

                    if (readLength > 0)
                    {
                        // join the new buffer to the end of the last message
                        message = String.Concat(message, encoder.GetString(bytes, 0, readLength));

                        // find the end of the next message
                        messageLength = message.IndexOf(TERMINATOR);

                        while (messageLength > 0)
                        {
                            // enqueue the message
                            m_MessageQueue.Enqueue(message.Substring(0, messageLength));

                            // signal the pump thread that there is a message waiting
                            m_ThreadReset.Set();

                            // remove message + the terminator charactor
                            message = message.Remove(0, messageLength + 1);

                            // find the end of the next message if it exists
                            messageLength = message.IndexOf(TERMINATOR);
                        }
                    }
                }
            }
            catch (IOException e)
            {
                if(m_ChiXMDClient.CurrentState != ChiXMDClient.States.Logout)
                    m_ChiXMDClient.ProcessState(ChiXMDClient.States.FailedToRead);

                // shutdown the client
                DeInitialise();
            }
            catch (Exception e)
            {
                // FOR DEBUGGING ONLY ?!?!?!
                if (!(e is ThreadAbortException))
                {
                    m_ChiXMDClient.ProcessState(ChiXMDClient.States.ThreadingIssue);

                    // shutdown the client
                    DeInitialise();
                }
            }

            
        }


        /// <summary>
        /// Reads from the message queue, decodes it into an IMessage object and sends out as an event
        /// </summary>
        private void PumpMessages()
        {
            string rawMessage;
            Messages.IMessage message = Messages.Heartbeat.GetInstance();

            while(m_ThreadRead.IsAlive)
            {
                while(m_MessageQueue.TryDequeue(out rawMessage))
                {
                    MessageParser.TryParseSequencedMessage(ref rawMessage, ref message);
                    m_ChiXMDClient.ProcessMessage(message);
                }

                m_ThreadReset.WaitOne();
            }
        }


    }
}
